package streaming

import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object DriverHA {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf()
    sparkConf.setAppName("DriverHA")
    sparkConf.setMaster("local[4]")
    val sc = new SparkContext(sparkConf)
    val checkpointPath = "data/driverha"

    def createContext(): StreamingContext = {
      println("*" * 100)
      println("create StreamingContext")
      println("*" * 100)
      val ssc = new StreamingContext(sc, Durations.seconds(5))
      ssc.checkpoint(checkpointPath)
      val ds = ssc.socketTextStream("master", 9999)
      ds.map((_, 1)).reduceByKey(_ + _).print()
      ssc
    }

    val ssc = StreamingContext.getOrCreate(checkpointPath, createContext)
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

    /**
     * spark-submit --master yarn-cluster --supervise --class className --executor-memory 512m  --total-executor-cores 2  yourJars
     */
  }

}
